package com.gome.ss.flowcrawl.extractor;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.log4j.Logger;

import com.gome.ss.common.AppConfiguer;
import com.gome.ss.common.Utils;
import com.mongodb.DBObject;

public class FlowCrawlExtractorTask implements FlowCrawlExtractorTaskInterface {
	private static Logger logger = Logger.getLogger(FlowCrawlExtractorTask.class);
	
	protected ConnectionFactory connectionFactory = null;
	protected Connection connection = null;
	protected Session session = null;
	protected Destination destination = null;
	protected MessageProducer producer = null;
	
	protected boolean needStop = false;
	protected FlowCrawlExtractor extractor = null;
	
	public FlowCrawlExtractorTask(FlowCrawlExtractor extractor) {
		this.init(extractor, false);
	}
	
	public FlowCrawlExtractorTask(FlowCrawlExtractor extractor, boolean doIndex) {
		this.init(extractor, doIndex);
	}
	
	private void init(FlowCrawlExtractor extractor, boolean doIndex) {
		this.extractor = extractor;
        connectionFactory = new ActiveMQConnectionFactory(
                ActiveMQConnection.DEFAULT_USER,
                ActiveMQConnection.DEFAULT_PASSWORD,
                AppConfiguer.getStringByKey("activeMQHost"));
        
        try {
			connection = connectionFactory.createConnection();
			connection.start();
			session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
			if (doIndex) {
				Destination destination = session.createQueue(FlowCrawlExtractorManager.esdqName);
				producer = session.createProducer(destination);
			} else {
				destination = session.createQueue(FlowCrawlExtractorManager.mqprefix + extractor.getConfig().getName() + "_" + extractor.getConfig().getStepID());
				producer = session.createProducer(destination);
			}
		} catch (Exception e) {
			logger.error(Utils.getTrace(e));
		}
	}
	
	protected void putQueue(DBObject obj) throws JMSException {
        TextMessage message = session.createTextMessage(obj.toString());
        producer.send(message);
	}
	
	@Override
	public void stop() {
		needStop = true;
	}

	protected void close() {
		try {
			producer.close();
			session.close();
			connection.close();
		} catch (JMSException e) {
			logger.error(Utils.getTrace(e));
		}
	}
}
